package com.mjf.day3

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
 * Flink写入到Redis
 */
object MyRedisSink {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\flink-stu\\src\\main\\input\\sensor.txt")

    val dataStream: DataStream[SensorReading] = inputStream.map(
      data => {
        val dataArr: Array[String] = data.split(",")
        SensorReading(dataArr(0), dataArr(1).toLong, dataArr(2).toDouble)
      }
    )

    // 定义一个redis的配置类
    val conf: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
      .setHost("hadoop103")
      .setPort(6379)
      .build()
    // 定义一个RedisMapper
    val myMapper: RedisMapper[SensorReading] = new RedisMapper[SensorReading] {
      // 定义保存数据到Redis的命令,type:hash hset table_name key value
      override def getCommandDescription: RedisCommandDescription = {
        new RedisCommandDescription(RedisCommand.HSET, "sensor_temp")
      }

      override def getKeyFromData(data: SensorReading): String = data.id

      override def getValueFromData(data: SensorReading): String = data.temperature.toString
    }

    // 数据保存到redis
    dataStream.addSink(
      new RedisSink(conf, myMapper)
    )

    env.execute("MyRedisSink")

  }
}
